package com.atguigu.flink.day03;

import com.atguigu.flink.bean.WaterSensor;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @author Felix
 * @date 2024/8/12
 * 该案例演示了转换算子-flatMap
 * 需求： 如果输入的数据是sensor_1，只打印vc；如果输入的数据是sensor_2，既打印ts又打印vc。
 */
public class Flink11_Transform_FlatMap {
    public static void main(String[] args) throws Exception {
        //TODO 1.指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //TODO 2.准备数据
        DataStreamSource<WaterSensor> stream = env.fromElements(
                new WaterSensor("sensor_1", 1L, 1),
                new WaterSensor("sensor_2", 2L, 2)
        );
        //TODO 3.map和flatMap
        //map：通过方法的返回值将数据传递到下游  ，返回值只能有一个，所以只会向下游传递一条数据
        //flatMap：返回值是void，不是通过返回值将数据传递到下游 ，它是通过collect方法向下游传递数据，collect方法可以调用多次
        /*stream.map(
                new MapFunction<WaterSensor, String>() {
                    @Override
                    public String map(WaterSensor ws) throws Exception {
                        if("sensor_1".equals(ws.id)){
                            return ws.vc + "";
                        }else {
                            return ws.vc + "";
                            return ws.ts + "";
                        }

                    }
                }
        )*/
        SingleOutputStreamOperator<String> ds = stream.flatMap(
                new FlatMapFunction<WaterSensor, String>() {
                    @Override
                    public void flatMap(WaterSensor ws, Collector<String> out) throws Exception {
                        if ("sensor_1".equals(ws.id)) {
                            out.collect(ws.vc + "");
                        } else {
                            out.collect(ws.vc + "");
                            out.collect(ws.ts + "");
                        }
                    }
                }
        );

        //TODO 4.打印
        ds.print();

        //TODO 5.提交作业
        env.execute();


    }
}
